package com.jh.paymentchannel.business.impl;

import cn.jh.common.core.json.JsonKt;
import cn.jh.common.utils.DateUtil;
import com.fasterxml.jackson.databind.type.TypeFactory;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Range;
import com.google.common.collect.Table;
import com.jh.paymentchannel.business.PaymentOrderStatisticBusiness;
import com.jh.paymentchannel.payment.OrderType;
import com.jh.paymentchannel.payment.PaymentAssemblers;
import com.jh.paymentchannel.payment.PosDTO;
import com.jh.paymentchannel.pojo.PaymentOrder;
import com.jh.paymentchannel.pojo.PaymentOrderStatistic;
import com.jh.paymentchannel.pojo.ShareProfitRate;
import com.jh.paymentchannel.pojo.ShareProfitRate.ProfitRatePart;
import com.jh.paymentchannel.pojo.UserTurnover;
import com.jh.paymentchannel.pojo.profit.PmtProfitRateProvider;
import com.jh.paymentchannel.repository.PaymentOrderRepository;
import com.jh.paymentchannel.repository.PaymentOrderStatisticRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConstructorArgumentValues;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.client.RestOperations;
import org.springframework.web.client.RestTemplate;

import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.temporal.ChronoField;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;


/**
 * @author Fuchun
 * @since 1.0
 */
@Service("paymentOrderStatisticBusiness")
public class PaymentOrderStatisticBusinessImpl implements PaymentOrderStatisticBusiness {

    @SuppressWarnings("unchecked")
    private static final Class<Map<String, Object>> MAP_CLASS =
            (Class<Map<String, Object>>) TypeFactory.defaultInstance()
                    .constructMapType(Map.class, String.class, Object.class).getRawClass();

    private static final BigDecimal DEFAULT_BOUNDARY = new BigDecimal("300");

    // PaymentOrder.totalFee <= DEFAULT_BOUNDARY
    private static final Specification<PaymentOrder> AT_MOST_BOUNDARY_SPEC =
            (root, query, cb) -> cb.le(root.get("totalFee"), DEFAULT_BOUNDARY);

    private static final Specification<PaymentOrder> GREATER_THAN_BOUNDARY_SPEC =
            (root, query, cb) -> cb.greaterThan(root.get("totalFee"), DEFAULT_BOUNDARY);

    private final Logger logger = LoggerFactory.getLogger(getClass());

    // 这里写死了分润费率，等待分润费率配置接口
    private final ShareProfitRate shareProfitRate = new ShareProfitRate.Builder()
            .addRateParts(
                    new ProfitRatePart(Range.atMost(DEFAULT_BOUNDARY), new BigDecimal("0.006")),
                    new ProfitRatePart(Range.greaterThan(DEFAULT_BOUNDARY), new BigDecimal("0.0022")))
            .build();
    private final PaymentOrderRepository paymentOrderRepository;
    private final PaymentOrderStatisticRepository paymentOrderStatisticRepository;
    private final RestOperations restOperations;
    private final PmtProfitRateProvider pmtProfitRateProvider;

    @Autowired
    public PaymentOrderStatisticBusinessImpl(
            PaymentOrderRepository paymentOrderRepository,
            PaymentOrderStatisticRepository paymentOrderStatisticRepository,
            @Qualifier("restOperations") RestOperations restOperations,
            PmtProfitRateProvider pmtProfitRateProvider) {
        this.paymentOrderRepository = paymentOrderRepository;
        this.paymentOrderStatisticRepository = paymentOrderStatisticRepository;
        this.restOperations = restOperations;
        this.pmtProfitRateProvider = pmtProfitRateProvider;
    }

    @Override
    @Transactional
    public Long statsLastWeekOrders(LocalDate statsDate) {
        Range<LocalDateTime> lastWeekDateRange = DateUtil.lastWeekDateTimeRange(statsDate);
        Range<Date> dateRange = DateUtil.toOldDateRange(lastWeekDateRange);
        List<Long> userIds = paymentOrderRepository.findUserIdsOnTradeSuccess(dateRange);
        Map<Long, UserTurnover> turnoversAtMost =
                paymentOrderRepository.findUserTurnovers(dateRange, AT_MOST_BOUNDARY_SPEC);
        Map<Long, UserTurnover> turnoversGreaterThan =
                paymentOrderRepository.findUserTurnovers(dateRange, GREATER_THAN_BOUNDARY_SPEC);

        List<PaymentOrderStatistic> posList = new ArrayList<>(userIds.size());

        for (Long userId : userIds) {
            UserTurnover atMost = turnoversAtMost.get(userId);
            UserTurnover greaterThan = turnoversGreaterThan.get(userId);
            if (atMost == null && greaterThan == null) {
                // 该用户没有产生实际的交易
                continue;
            }

            // Bug: old implements NOT SUPPORT orderType
            PaymentOrderStatistic pos = new PaymentOrderStatistic(userId, OrderType.ALIPAY);
            pos.setDealNumber(0);
            pos.setTotalProfit(BigDecimal.ZERO);
            pos.setTotalAmount(BigDecimal.ZERO);
            pos.setRateProps(JsonKt.toJSONString(shareProfitRate));
            pos.setFromDate(lastWeekDateRange.lowerEndpoint().toLocalDate());
            pos.setToDate(lastWeekDateRange.upperEndpoint().toLocalDate());
            pos.setWeekOfYear(lastWeekDateRange.lowerEndpoint().get(ChronoField.ALIGNED_WEEK_OF_YEAR));
            pos.setTransferred(false);
            pos.setCreateTime(LocalDateTime.now());

            if (atMost != null)
                pos.addAmountAndProfit(atMost, shareProfitRate.getRate(atMost.getTotalAmount()));
            if (greaterThan != null) {
                pos.addAmountAndProfit(greaterThan, shareProfitRate.getRate(greaterThan.getTotalAmount()));
            }

            posList.add(pos);
        }

        return paymentOrderStatisticRepository.saveAndFlush(posList);
    }

    @Override
    @Transactional
    public Page<Long> statsLastWeekOrders(LocalDate statsDate, Pageable pageable) {
        return statsLastWeekOrders(statsDate, pageable, null);
    }

    Page<Long> statsLastWeekOrders(LocalDate statsDate, Pageable pageable, Consumer<List<PaymentOrderStatistic>> consumer) {
        Range<LocalDateTime> lastWeekDateRange = DateUtil.lastWeekDateTimeRange(statsDate);
        Range<Date> dateRange = DateUtil.toOldDateRange(lastWeekDateRange);

        long posRecords = paymentOrderStatisticRepository.countByFromDateAndToDate(
                lastWeekDateRange.lowerEndpoint().toLocalDate(),
                lastWeekDateRange.upperEndpoint().toLocalDate());
        if (posRecords > 0) {
            if (logger.isDebugEnabled()) {
                logger.debug("本周订单数据已完成分润统计计算。");
            }
            return new PageImpl<>(Collections.emptyList(), pageable, 0L);
        }

        Page<PaymentOrder> poPage = paymentOrderRepository.findTradingSuccess(dateRange, null, pageable);
        if (!poPage.hasContent()) {
            if (logger.isDebugEnabled()) {
                logger.debug("本周（{} - {}）订单交易成功笔数为0，不计算分润。",
                        dateRange.lowerEndpoint(),
                        dateRange.upperEndpoint());
            }
            return new PageImpl<>(Collections.emptyList(), pageable, 0L);
        }
        Table<Long, OrderType, PaymentOrderStatistic> posTable = HashBasedTable.create(
                poPage.getNumberOfElements(), OrderType.values().length);
        for (PaymentOrder order : poPage.getContent()) {
            PaymentOrderStatistic pos = posTable.get(order.getUserId(), order.getOrderType());
            if (pos == null) {
                pos = new PaymentOrderStatistic(order.getUserId(), order.getOrderType()).init();
                pos.setRateProps(pmtProfitRateProvider.getProfitRateJson());
                pos.setFromDate(lastWeekDateRange.lowerEndpoint().toLocalDate());
                pos.setToDate(lastWeekDateRange.upperEndpoint().toLocalDate());
                pos.setWeekOfYear(lastWeekDateRange.lowerEndpoint().get(ChronoField.ALIGNED_WEEK_OF_YEAR));
                pos.setCreateTime(LocalDateTime.now());

                posTable.put(order.getUserId(), order.getOrderType(), pos);
            }

            BigDecimal rate = pmtProfitRateProvider.getProfitRate()
                    .getRate(order.getOrderType(), order.getTotalFee());

            pos.addAmountAndProfit(order, rate);
        }
        List<PaymentOrderStatistic> posList = new ArrayList<>(posTable.values());
        posList.sort(Comparator.comparing(PaymentOrderStatistic::getUserId));
        Long rows = paymentOrderStatisticRepository.saveAndFlush(posList);
        if (logger.isDebugEnabled()) {
            logger.debug("Add new PaymentOrderStatistic rows: {}", rows);
        }
        if (consumer != null) {
            consumer.accept(posList);
        }
        List<Long> userIds = new ArrayList<>(posTable.size());
        return new PageImpl<>(userIds, pageable, poPage.getTotalElements());
    }

    @Override
    @Transactional(readOnly = true)
    public Map<Long, Boolean> verifySyncPosIds(Collection<Long> posIds) {
        return paymentOrderStatisticRepository.findIdAndTransferred(posIds);
    }

    @Override
    @Transactional
    public Long syncPosToUser(Pageable pageable) {
        Page<PaymentOrderStatistic> posPage = paymentOrderStatisticRepository.findByTransferredFalse(pageable);
        if (!posPage.hasContent()) {
            return 0L;
        }
        Long syncCount = syncPaymentProfits(posPage.getContent());
        if (syncCount <= 0) {
            return syncCount;
        }
        List<Long> posIds = posPage.getContent().stream()
                .map(PaymentOrderStatistic::getId).collect(Collectors.toList());
        return paymentOrderStatisticRepository.setTransferred(posIds);
    }

    private HttpEntity<List<PosDTO>> requestEntity(List<PosDTO> dtoList) {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
        headers.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
        return new HttpEntity<>(dtoList, headers);
    }

    // 0:  No content 不需要同步
    // -1: 请求返回失败 (user module error)
    // -2: 请求时发生错误  (payment module error)
    private Long syncPaymentProfits(List<PaymentOrderStatistic> content) {
        if (content == null || content.isEmpty()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Sync PaymentOrderStatistic data is empty.");
            }
            return 0L;
        }
        List<PosDTO> dtoList = new ArrayList<>(content.size());
        for (PaymentOrderStatistic pos : content) {
            dtoList.add(PaymentAssemblers.toDto(pos));
        }

        if (logger.isDebugEnabled()) {
            logger.debug("Sync PaymentOrderStatistic data to user.PaymentProfit: {} records", dtoList.size());
        }
        try {
            ResponseEntity<Map<String, Object>> response = restOperations.postForEntity(
                    "http://user/v1.0/user/payment_profits/sync", requestEntity(dtoList), MAP_CLASS);
            HttpStatus statusCode = response.getStatusCode();
            if (statusCode.is2xxSuccessful()) {
                switch (statusCode) {
                    case OK:
                        Map<String, Object> respMap = response.getBody();
                        if (respMap == null || respMap.isEmpty()) {
                            return 0L;
                        }
                        long syncCount = ((Number) respMap.getOrDefault("syncCount", 0L)).longValue();
                        logger.debug("sync payment profits syncCount = {}", syncCount);
                        return syncCount;
                    default:
                        return 0L;
                }
            } else {
                return -1L;
            }
        } catch (Exception ex) {
            String message = ex.getLocalizedMessage() != null ? ex.getLocalizedMessage() : ex.getMessage();
            if (message == null) {
                Throwable cause = ex.getCause();
                message = cause == null ? ex.getClass().getName() : cause.getMessage();
            }
            // Net exception / IOException
            logger.error("/v1.0/user/payment_profits/sync exception: {}", message);
            return -2L;
        }
    }
}
